package com.bw.app.dwm;

import com.alibaba.fastjson.JSON;
import com.bw.bean.OrderWideNew;
import com.bw.bean.PaymentInfo;
import com.bw.bean.PaymentWideNew1;
import com.bw.utils.ClickHouseUtil;
import com.bw.utils.DateTimeUtil;
import com.bw.utils.MyKafkaUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.ProcessJoinFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.util.Collector;

import java.time.Duration;

/**
 * @ProjectName: BigData
 * @Package: com.bw.app.dwm
 * @ClassName: PaymentWideApp
 * @Author: Gy
 * @Description: 支付宽表处理程序
 * @Date: 2021/11/11 18:39
 */
public class PaymentWideApp {
    public static void main(String[] args) throws Exception {
        //todo 1.基本环境的准备
        //1.1 创建流式处理环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //1.2 设置并行度
        env.setParallelism(1);

        //todo 2.从kafka的主题中读取数据
        //2.1 声明相关的主题以及消费者组
        String paymentInfoSourceTopic = "dwd_payment_info";
        String orderWideSourceTopic = "dwm_order_wide";
        String paymentWideSinkTopic = "dwm_payment_wide";
        String groupId = "paymentwide_app_group";

        //2.1读取支付数据
        FlinkKafkaConsumer<String> paymentInfoSource = MyKafkaUtil.getKafkaSource(paymentInfoSourceTopic, groupId);
        DataStreamSource<String> paymentInfoJsonStrDS = env.addSource(paymentInfoSource);

        //2.2读取订单宽表数据
        FlinkKafkaConsumer<String> orderWideSource = MyKafkaUtil.getKafkaSource(orderWideSourceTopic, groupId);
        DataStreamSource<String> orderWideJsonStrDS = env.addSource(orderWideSource);

        //todo 3.对读取到的数据进行结构的转换     jsonStr->POJO
        //3.1 转换支付流
        SingleOutputStreamOperator<PaymentInfo> paymentInfoDS = paymentInfoJsonStrDS.map(
                jsonStr -> JSON.parseObject(jsonStr, PaymentInfo.class)
        );
        //3.2 转换订单流
        SingleOutputStreamOperator<OrderWideNew> orderWideDS = orderWideJsonStrDS.map(
                jsonStr -> JSON.parseObject(jsonStr, OrderWideNew.class)
        );

        //todo 4.设置Watermark以及提取事件时间段
        //4.1 支付流的Watermark
        SingleOutputStreamOperator<PaymentInfo> paymentInfoWithWatermarkDS = paymentInfoDS.assignTimestampsAndWatermarks(
                WatermarkStrategy
                        .<PaymentInfo>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                        .withTimestampAssigner(new SerializableTimestampAssigner<PaymentInfo>() {
                            @Override
                            public long extractTimestamp(PaymentInfo paymentInfo, long l) {
                                return DateTimeUtil.toTs(paymentInfo.getCallback_time());
                            }
                        })
        );

        //4.2订单流的Watermark
        SingleOutputStreamOperator<OrderWideNew> orderWideWithWatermarkDS = orderWideDS.assignTimestampsAndWatermarks(
                WatermarkStrategy
                        .<OrderWideNew>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                        .withTimestampAssigner(new SerializableTimestampAssigner<OrderWideNew>() {
                            @Override
                            public long extractTimestamp(OrderWideNew orderWideNew, long l) {
                                return DateTimeUtil.toTs(orderWideNew.getCreate_time());
                            }
                        })
        );

        //todo 5.对数据进行分组
        //5.1 支付数据分组
        KeyedStream<PaymentInfo, Long> paymentInfoKeyedDS = paymentInfoWithWatermarkDS.keyBy(PaymentInfo::getOrder_id);
        //5.2 订单宽表流数据分组
        KeyedStream<OrderWideNew, Long> orderWideKeyedDS = orderWideWithWatermarkDS.keyBy(OrderWideNew::getOrder_id);

        //todo 6.使用intervalJoin对订单和订单明细进行关联
        SingleOutputStreamOperator<PaymentWideNew1> paymentWideDS = paymentInfoKeyedDS.intervalJoin(orderWideKeyedDS)
                .between(Time.seconds(-1800), Time.seconds(0))
                .process(new ProcessJoinFunction<PaymentInfo, OrderWideNew, PaymentWideNew1>() {
                    @Override
                    public void processElement(PaymentInfo paymentInfo, OrderWideNew orderWideNew, Context context, Collector<PaymentWideNew1> collector) throws Exception {
                        collector.collect(new PaymentWideNew1(paymentInfo, orderWideNew));
                    }
                });
        paymentWideDS.print(">>>>>");

        //todo 7.将数据写到kafka的dwm层
        paymentWideDS.map(
                paymentwide -> JSON.toJSONString(paymentwide)
        ).addSink(MyKafkaUtil.getKafkaSink(paymentWideSinkTopic))
        ;
        //将对象转换为JSONObject 输出到clickhouse作为用户画像的基础分析数据
        paymentWideDS.addSink(
                ClickHouseUtil.getJdbcSink(" insert into protrait_payment(category1_id,category2_id,category3_id,category1_name,category2_name, category3_name,carrier,carriername, user_gendertype,user_gender,province_id,province_name,tm_name,email, emailtype,user_age,yearbasetype,yearbasename,payment_create_time,user_id,order_id,order_price,sku_num ,coupon_reduce_amount,total_amount,activity_reduce_amount,payment_total_amount,spu_name) values(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)")
        );

        env.execute();
    }
}
